package com.heima.kafka.sample;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.*;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Properties;

/**
 * @author brianxia
 * @version 1.0
 * @date 2022-10-31 9:50
 */
public class KafkaStreamQuickStart107 {

    public static void main(String[] args) {
        //创建配置属性
        Properties properties = new Properties();
        properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");
        //对key value序列化器进行声明 字符串
        properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        //对应用id进行唯一赋值
        properties.put(StreamsConfig.APPLICATION_ID_CONFIG,"teststreams107");


        //设计流式计算
        StreamsBuilder streamsBuilder = new StreamsBuilder();

        //获取数据，指定topic名字
        KStream<String, String> stream = streamsBuilder.stream("streams-test-107");
        //计算方法写法和stream流处理基本一致
        //value "hello kafka"  -> [value : "hello",value : "kakfa"]
        stream.flatMapValues(value -> Arrays.asList(value.split(" ")))
                //按value分组
                .groupBy((key, value) -> value)
                //设置统计时间窗口->每隔多少时间统计一次
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
                //key : "hello"  value："hello" -> count
                .count()
                //key : hello value: 2 将KTABLE转换成KSTREAM
                .toStream()
                //将key value调整成String类型
                .map((key, value) -> new KeyValue<>(key.key(), value.toString()))
                //投递结果到kafka
                .to("streams-out-107");


        //创建对象
        KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(),properties);

        kafkaStreams.start();

    }
}
